package com.thp.bigdata.webClick.mr;

import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.Date;
import java.util.Locale;
import java.util.UUID;

import org.apache.commons.beanutils.BeanUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import com.thp.bigdata.webClick.mrBean.WeblogBean;

/**
 * 将清洗之后的日志梳理出点击流pageViews模型数据
 * 输入的数据是经过清洗之后的数据
 * 
 * 区分每一次会话，给每一次visit(session)增加了session-id(随机uuid)
 * 梳理出每一次会话中所访问的每个页面（请求时间，url，停留时长，以及该页面在这次session中的序号）
 * 保留referral_url，body_bytes_send，useragent
 * @author 汤小萌
 *
 */
public class ClickStream {
	
	static class ClickStreamMapper extends Mapper<LongWritable, Text, Text, WeblogBean> {
		Text k = new Text();
		WeblogBean v = new WeblogBean();
		
		@Override
		protected void map(LongWritable key, Text value, Context context)
				throws IOException, InterruptedException {
			String line = value.toString();
System.out.println(line);
			String[] fields = line.split("\001");
			if(fields.length < 9) return;
			v.set("true".equals(fields[0]) ? true : false, fields[1], fields[2], fields[3], fields[4], fields[5], 
					fields[6], fields[7], fields[8]);
			// 只有有效的记录才会进入后续处理
			if(v.isValid()) {
				k.set(v.getRemote_addr());
				context.write(k, v);
			}
		}
	}
	
	static class ClickStreamReducer extends Reducer<Text, WeblogBean, NullWritable, Text> {
		Text v = new Text();
		@Override
		protected void reduce(Text key, Iterable<WeblogBean> values, Context context) throws IOException, InterruptedException {
			ArrayList<WeblogBean> beans = new ArrayList<WeblogBean>();
			// 先将每一个用户都拿出来按照时间进行排序
			for(WeblogBean bean : values) {
				WeblogBean weblogBean = new WeblogBean();
				try {
					BeanUtils.copyProperties(weblogBean, bean);
				} catch (IllegalAccessException | InvocationTargetException e) {
					e.printStackTrace();
				}
				beans.add(weblogBean);
			}
			
			
			// 将Bean按照时间进行排序
			Collections.sort(beans, new Comparator<WeblogBean>() {
				@Override
				public int compare(WeblogBean o1, WeblogBean o2) {
					try {
						Date d1 = toDate(o1.getTime_local());
						Date d2 = toDate(o2.getTime_local());
						if(d1 == null || d2 == null) return 0;
						return d1.compareTo(d2);
					} catch (ParseException e) {
						e.printStackTrace();
					}
					return 0;
				}
				
			});
			
			int step = 1;
			String session = UUID.randomUUID().toString();
			for(int i = 0; i < beans.size(); i++) {
				WeblogBean bean = beans.get(i);
				// 如果仅有一条数据，则输出
				if(1 == beans.size()) {
					// 设置默认停留时间为60s
					v.set(session+"\001"+key.toString()+"\001"+bean.getRemote_user() + "\001" +
					bean.getTime_local() + "\001" + bean.getRequest() + "\001" + step + "\001" + (60) + 
					"\001" + bean.getHttp_referer() + "\001" + bean.getHttp_user_agent() + "\001" + 
					bean.getBody_bytes_sent() + "\001"
							+ bean.getStatus());
					context.write(NullWritable.get(), v);
					// 重新生成session
					session = UUID.randomUUID().toString();
					break;
				} 
				if(i == 0) {  // 不止一条数据，那么第一条要直接跳过，因为 bean.get(i-1)
					continue;
				}
				
				try {
					long timeDiff = timeDiff(bean.getTime_local(), beans.get(i - 1).getTime_local());
					if(timeDiff < 30*60*1000) {
						// 如果  本次  -  上次   时间差   <  30  min  ，则输出前一次的页面访问信息
						v.set(session+"\001"+key.toString()+"\001"+beans.get(i - 1).getRemote_user() + 
								"\001" + beans.get(i - 1).getTime_local() + "\001" + beans.get(i - 1).getRequest() + 
								"\001" + step + "\001" + (timeDiff / 1000) + "\001" + beans.get(i - 1).getHttp_referer() + "\001"
								+ beans.get(i - 1).getHttp_user_agent() + "\001" + beans.get(i - 1).getBody_bytes_sent() + "\001"
								+ beans.get(i - 1).getStatus());
						context.write(NullWritable.get(), v);
						step++;
					} else {
						// 如果  本次 - 上次 时间差  > 30 min, 则输出前一次的页面访问信息，将step重置为1，以分隔为为新的visit
						v.set(session+"\001"+key.toString()+"\001"+beans.get(i - 1).getRemote_user() + "\001" +
						beans.get(i - 1).getTime_local() + "\001" + beans.get(i - 1).getRequest() + "\001" + 
								(step) + "\001" + (60) + "\001" + beans.get(i - 1).getHttp_referer() + "\001"
								+ beans.get(i - 1).getHttp_user_agent() + "\001" + 
								beans.get(i - 1).getBody_bytes_sent() + "\001" + beans.get(i - 1).getStatus());
						context.write(NullWritable.get(), v);
						// 输出完上一条之后，重置step编号
						step = 1;
						// session 也要重新生成
						session = UUID.randomUUID().toString();
					}
				} catch (ParseException e) {
					e.printStackTrace();
				}
				
				// 如果此次遍历时最后一条数据，则将本条数据输出  session  在上面的逻辑都控制了
				if(i == beans.size() - 1) {
					// 设置默认停留时间为60s
					v.set(session+"\001"+key.toString()+"\001"+bean.getRemote_user() + "\001" + bean.getTime_local() + "\001" + bean.getRequest() + "\001" + step + "\001" + (60) + "\001" + bean.getHttp_referer() + "\001" + bean.getHttp_user_agent() + "\001" + bean.getBody_bytes_sent() + "\001" + bean.getStatus());
					context.write(NullWritable.get(), v);
				}
				
				
				
			}
			
			
			
			
		}
		
		
		private String toStr(Date date) {
			SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss",Locale.US);
			return sdf.format(date);
		}
		
		private Date toDate(String timeStr) throws ParseException {
			SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss",Locale.UK);
			return sdf.parse(timeStr);
		}
		
		// 算时间差
		private long timeDiff(String time1, String time2) throws ParseException {
			Date d1 = toDate(time1);
			Date d2 = toDate(time2);
			return d1.getTime() - d2.getTime();
		}
		
	}
	
	public static void main(String[] args) throws IllegalArgumentException, IOException, ClassNotFoundException, InterruptedException {
		Configuration conf = new Configuration();
		Job job = Job.getInstance(conf);

		job.setJarByClass(ClickStream.class);

		job.setMapperClass(ClickStreamMapper.class);
		job.setReducerClass(ClickStreamReducer.class);

		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(WeblogBean.class);

		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(Text.class);

//		 FileInputFormat.setInputPaths(job, new Path(args[0]));
//		 FileOutputFormat.setOutputPath(job, new Path(args[1]));

		FileInputFormat.setInputPaths(job, new Path("f:/weblog_1/output"));
		FileOutputFormat.setOutputPath(job, new Path("f:/weblog_1/pageviews"));
		
		FileSystem fs = FileSystem.get(conf);
		if(fs.exists(new Path("f:/weblog_1/pageviews"))) {
			fs.delete(new Path("f:/weblog_1/pageviews"), true);
		}

		job.waitForCompletion(true);
	}
	
	
	
	
}
